package org.ws.task.netty.support;

import com.google.common.collect.Maps;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.common.AddressUtils;
import org.ws.task.netty.handler.ActionHandlerRegistry;
import org.ws.task.netty.NettyConfig;
import org.ws.task.netty.RequestActionHandler;
import org.ws.task.model.protocol.base.RequestAction;
import org.ws.task.model.protocol.base.RequestMessage;
import org.ws.task.model.protocol.base.ResponseMessage;
import org.ws.task.model.protocol.base.ResponseStatus;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

public abstract class AbstractNettyService {

    private static Logger log = LoggerFactory.getLogger(AbstractNettyService.class.getName());

    protected static final int SO_SNDBUF = 1024 * 32;
    protected static final int SO_RCVBUF = 1024 * 32;

    protected final NettyConfig nettyConfig;

    private final Map<Integer,ResponseFuture> responseFutureMap = Maps.newConcurrentMap();

    private ExecutorService actionHandlerExecutor;
    private ActionHandlerRegistry actionHandlerRegistry;

    private final Semaphore asyncSemaphore;
    private final Semaphore oneWaySemaphore;

    public AbstractNettyService(NettyConfig nettyConfig) {
        this.nettyConfig = nettyConfig;
        this.asyncSemaphore = new Semaphore(this.nettyConfig.getRequestThreadPermitsCount(),true);
        this.oneWaySemaphore = new Semaphore(this.nettyConfig.getRequestThreadPermitsCount(),true);
    }


    protected abstract ExecutorService getCallbackExecutor();

    protected void setActionHandlerRegistry(ActionHandlerRegistry actionHandlerRegistry, ExecutorService executor) {
        this.actionHandlerExecutor = executor;
        this.actionHandlerRegistry = actionHandlerRegistry;
    }

    protected ResponseMessage doSync(Channel channel, RequestMessage request, long timeout)throws Exception{
        if(isChannelOk(channel)){
            final int requestId = request.getId();
            ResponseFuture responseFuture = new ResponseFuture(channel,request,timeout,null,(future)->{
                log.debug("doSync callback:{}",requestId);
            });
            AbstractNettyService.this.responseFutureMap.put(requestId,responseFuture);
            writeAndFlush(channel,request,responseFuture);
            if (!responseFuture.syncAwait()) {
                throw new TimeoutException("doSync timeout requestId:"+requestId);
            }
            ResponseMessage response = responseFuture.getResponse();
            if(response != null){
                return response;
            }
            else {
                // error
                throw new Exception("doSync response is null.",responseFuture.getCause());
            }
        }
        else {
            // channel error
            throw new Exception("doSync channel is not ok.");
        }
    }

    protected void doAsync(final Channel channel, final RequestMessage request, final long timeout,
                           final Consumer<ResponseFuture> callback)throws Exception{
        if(isChannelOk(channel)){
            final int requestId = request.getId();
            if(this.asyncSemaphore.tryAcquire(timeout, TimeUnit.MILLISECONDS)){
                ResponseFuture responseFuture = new ResponseFuture(channel,request,timeout,this.asyncSemaphore,callback);
                AbstractNettyService.this.responseFutureMap.put(requestId,responseFuture);
                writeAndFlush(channel,request,responseFuture);
            }
            else {
                // timeout
                throw new TimeoutException("doAsync timeout requestId:"+requestId);
            }
        }
        else {
            // channel error
            throw new Exception("doAsync channel[{"+AddressUtils.getChannelRemoteAddress(channel)+"}] is not ok.");
        }
    }

    protected void doOneWay(Channel channel, RequestMessage request, long timeout)throws Exception{
        if(isChannelOk(channel)){
            final int requestId = request.getId();
            if(this.oneWaySemaphore.tryAcquire(timeout,TimeUnit.MICROSECONDS)){
                ResponseFuture responseFuture = new ResponseFuture(channel,request,timeout,this.oneWaySemaphore,future -> {
                    log.debug("doOneWay callback:{}",requestId);
                });
                writeAndFlush(channel,request,responseFuture);
                // no response
                responseFuture.executeCallback(getCallbackExecutor());
            }
            else {
                throw new TimeoutException("doOneWay timeout requestId:"+requestId);
            }
        }
        else {
            // channel error
            throw new Exception("doOneWay channel is not ok.");
        }
    }

    private void writeAndFlush(final Channel channel,final RequestMessage request,final ResponseFuture responseFuture){
        try {
            channel.writeAndFlush(request).addListener((future)->{
                if(!future.isSuccess()){
                    responseFuture.setCause(future.cause());
                    responseFuture.setResponse(null);
                    responseFuture.setSendSuccess(false);
                    responseFuture.executeCallback(getCallbackExecutor());
                    log.error("request[{}] send failed.",request.getId());
                }
                else {
                    responseFuture.setSendSuccess(true);
                    log.debug("request[{}] send success.",request.getId());
                }
            });
        } catch (Exception e) {
            responseFuture.setCause(e);
            responseFuture.setResponse(null);
            responseFuture.setSendSuccess(false);
            responseFuture.executeCallback(getCallbackExecutor());
            log.error(e.getMessage(),e);
        }
    }

    protected boolean isChannelOk(Channel channel){
        return channel != null && channel.isActive();
    }

    /**
     * 处理响应
     * @param ctx
     * @param response
     * @throws Exception
     */
    protected void processResponse(ChannelHandlerContext ctx, ResponseMessage response) throws Exception {
        int id = response.getId();
        ResponseFuture responseFuture = AbstractNettyService.this.responseFutureMap.get(id);
        if(responseFuture == null){
            log.error("request[{}] mapping responseFuture is null",id);
            return;
        }
        AbstractNettyService.this.responseFutureMap.remove(id);
        responseFuture.setResponse(response);
        responseFuture.executeCallback(getCallbackExecutor());
    }

    /**
     * 处理请求
     * @param ctx
     * @param request
     * @throws Exception
     */
    protected void processRequest(ChannelHandlerContext ctx, RequestMessage request){
        if(AbstractNettyService.this.actionHandlerExecutor == null ||
                AbstractNettyService.this.actionHandlerRegistry == null){
            return;
        }
        AbstractNettyService.this.actionHandlerExecutor.submit(()->{
            RequestActionHandler actionHandler = AbstractNettyService.this.actionHandlerRegistry.get(RequestAction.valueOf(request.getAction()));
            ResponseMessage response = null;
            final int requestId = request.getId();
            final int action = request.getAction();
            if(!RequestAction.valueOf(action).isAnswer()){
                return;
            }
            log.debug("request:{}",request);
            if(actionHandler != null){
                try {
                    response = actionHandler.handler(ctx, request);
                    response.setId(requestId);
                    response.setSuccess(true);
                    response.setStatus(ResponseStatus.SUCCESS.getCode());
                } catch (Exception e) {
                    response = new ResponseMessage(requestId);
                    response.setSuccess(false);
                    response.setStatus(ResponseStatus.ERROR.getCode());
                    response.setMessage(e.getMessage());
                    log.error(e.getMessage(),e);
                }
            }
            else {
                response = new ResponseMessage(requestId);
                response.setSuccess(false);
                response.setStatus(ResponseStatus.NO_HANDLER.getCode());
                log.error("action cannot find handler:{}",request.getAction());
            }
            //
            Channel channel = ctx.channel();
            channel.writeAndFlush(response).addListener((channelFuture)->{
                if(!channelFuture.isSuccess()){
                    log.error("send response[{}] to channel {} failed.", requestId, AddressUtils.getChannelRemoteAddress(channel),channelFuture.cause());
                }
                else {
                    log.debug("send response[{}] to channel {} success.", requestId, AddressUtils.getChannelRemoteAddress(channel));
                }
            });
        });
    }

    /**
     * 处理响应
     */
    class ResponseMessageHandler extends SimpleChannelInboundHandler<ResponseMessage> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ResponseMessage response) throws Exception {
            // 处理Server响应
            processResponse(ctx,response);
        }
    }

    /**
     * 处理请求
     */
    class RequestMessageHandler extends SimpleChannelInboundHandler<RequestMessage> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RequestMessage request) throws Exception {
            // 处理Leader请求
            processRequest(ctx,request);
        }
    }
}
